package test.cluster.net.connector;

import java.net.InetSocketAddress;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.serialization.ObjectSerializationCodecFactory;
import org.apache.mina.filter.codec.textline.TextLineCodecFactory;
import org.apache.mina.transport.socket.SocketConnector;
import org.apache.mina.transport.socket.nio.NioSocketConnector;

import test.cluster.MessagePacket;
import test.cluster.MessagePacket.Type;
import test.configuration.ConfigFactory;
import test.configuration.Configuration;
import test.domain.environment.node.Node;

public class MinaConnector implements Connector {
	private static final int CONNECTION_TIMEOUT = 5000;
	
	private Log log = LogFactory.getLog(this.getClass());
	private List<Node> nodes;
	
	List<IoSession> sessions = new ArrayList<IoSession>();

	@Override
	public Connector initialize(List<Node> nodes) {
		if (nodes == null)
			throw new IllegalArgumentException("nodes should not be null.");
		
		this.nodes = nodes;
		return this;
	}
	
	public void connect() {
		if (nodes == null || nodes.size() == 0)
			throw new IllegalStateException("Conector should be initialized.");
		
		SocketConnector connector = new NioSocketConnector();
		
		// configure the service
		connector.setConnectTimeoutMillis(CONNECTION_TIMEOUT);
//		connector.getFilterChain().addLast("codec", new ProtocolCodecFilter(new TextLineCodecFactory(Charset.forName("UTF-8"))));
		connector.getFilterChain().addLast("codec", new ProtocolCodecFilter(new ObjectSerializationCodecFactory()));
		connector.setHandler(new ConnectorSessionHandler());
		
		log.info("connect to the clients...");
		for (Node node : nodes) {
			ConnectFuture future = connector.connect(
					new InetSocketAddress(node.getIp(), node.getPort()));
			
			log.info("connecting " + node.getIp() + ":" + node.getPort());
			
			future.awaitUninterruptibly();
			IoSession session = future.getSession();
			session.setAttribute("ip", node.getIp());
			session.setAttribute("connector", connector);
			this.sessions.add(session);
		}
		
		log.info("connected to " + nodes.size() + " clients.");
	}
	
	public void send(Type message) {
		this.send(message.toString());
	}
	
	public void send(Object entity) {
		if (sessions == null || sessions.size() == 0)
			throw new IllegalStateException("connection should be established.");
		
		for (IoSession session : sessions) {
			log.info("sending message to clients: " + session.getAttribute("ip"));
			session.write(entity);
		}
	}
	
	public void disconnect() {
		if (sessions == null || sessions.size() == 0)
			throw new IllegalStateException("connection is null.");
		
		log.info("disconnecting...");
		for (IoSession session : sessions) {
			session.close(true);
			session = null;
		}
		log.info(sessions.size() + " sessions disconnected.");
	}
	
	public static void main(String[] args) {
		Node node = new Node();
		node.setIp("127.0.0.1");
		node.setPort(9000);
		
		List<Node> nodes = new ArrayList<Node>();
		nodes.add(node);
		
		Configuration config = ConfigFactory.getConfiguration();
		MessagePacket packet = new MessagePacket();
		
		packet.setTestClass(config.getTestClass());
		packet.setMetaThread(config.getMetaThread());
		packet.setMessage(MessagePacket.Type.getresult);
		
		Connector connector = new MinaConnector();
		connector.initialize(nodes).connect();
		connector.send(packet);
		connector.disconnect();		
	}
}
